package com.yanggu.flink.datastream_api.process_function.timerservice

import com.yanggu.flink.datastream_api.pojo.Event
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

//数据到达，时间戳为：1000, 水位线为：-9223372036854775808
//Flink程序开始时, 会发送一个Long.MIN_VALUE -1的一个Watermark，保证数据的Watermark一定向下游传递
//那么这里时间戳为1000的数据, Watermark为什么不是999呢？Watermark是每隔200ms生成一个, 第一个数据的到来时, Watermark并没有生成
//数据到达，时间戳为：11000, 水位线为：999
//这里数据之间间隔了5s, 因此Watermark周期性地生成了, 那么这里为什么不是10999而是999呢
//数据到达，时间戳为：11001, 水位线为：10999
//数据到达，时间戳为：11001, 水位线为：10999
//定时器触发，触发时间：11000, 水位线为: 11000
//这里由于Watermark更新了且大于等于定时器的时间戳, 定时器被执行了
//数据到达，时间戳为：1001, 水位线为：11000
//定时器触发，触发时间：11001, 水位线为: 9223372036854775807
//由于Source数据源是一个有界的数据流, 在数据流结束或者程序结束的时候, 会发送Long.MAX_VALUE的Watermark, 保证所有定时器和窗口都执行
//Flink定时器底层是一个优先队列, 按照时间戳升序排序, 队首是时间戳最小的元素。所以即使后添加也会先触发
//定时器触发，触发时间：21000, 水位线为: 9223372036854775807
//定时器触发，触发时间：21001, 水位线为: 9223372036854775807
object EventTimeTimerTest {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    env.addSource(new CustomSource2)
      .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner[Event] {
        override def extractTimestamp(element: Event, recordTimestamp: Long) = element.timestamp
      }))
      .keyBy(_ => true)
      .process(new KeyedProcessFunction[Boolean, Event, String] {
        override def processElement(value: Event, ctx: KeyedProcessFunction[Boolean, Event, String]#Context, out: Collector[String]) = {
          out.collect(s"数据到达，时间戳为：${ctx.timestamp}, 水位线为：${ctx.timerService().currentWatermark()}")
          //注册一个10s后的定时器
          //ctx.timestamp只会返回当前数据的时间戳，本质上调用withTimestampAssigner传入的接口的extractTimestamp方法
          ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 10 * 1000)
        }

        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Boolean, Event, String]#OnTimerContext, out: Collector[String]) = {
          out.collect(s"定时器触发，触发时间：$timestamp, 水位线为: ${ctx.timerService().currentWatermark()}")
        }
      })
      .print()

    env.execute()

  }

}

class CustomSource2 extends SourceFunction[Event] {

  def run(ctx: SourceContext[Event]): Unit = {
    // 直接发出测试数据
    ctx.collect(Event("Mary", "./home", 1000L))
    // 为了更加明显，中间停顿 5 秒钟
    Thread.sleep(5000L)
    // 发出 10 秒后的数据
    ctx.collect(Event("Mary", "./home", 11000L))
    Thread.sleep(5000L)
    // 发出 10 秒+1ms 后的数据
    ctx.collect(Event("Alice", "./cart", 11001L))
    // 发送时间戳一样的数据, 定时器会自动进行去重
    ctx.collect(Event("Alice", "./cart", 11001L))
    Thread.sleep(5000L)
    // 迟到的数据，并不会触发定时器的计算。只有当watermark发生更新时，才会触发定时器的计算
    ctx.collect(Event("Alice", "./cart", 1001L))
  }

  def cancel(): Unit = {
  }
  
}


